package main;

import helper.CompositeKey;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class SortReducer extends Reducer<CompositeKey, IntWritable, Text, Text> {
    
    
    
    public void reduce(CompositeKey key, Iterable<IntWritable> values, Context context) 
            throws IOException, InterruptedException {
        
//        StringBuffer sb = new StringBuffer("");
//        for (IntWritable val: values) {
//            sb.append(key.similar + "-" + val.get() + ", ");
//        }
        Map<Integer, Integer> similarMap = new HashMap<Integer, Integer>();
        
        for (IntWritable val : values) {
            int similar = key.getSimilar();
            if(similarMap.containsKey(similar)) {
                similarMap.put(similar, similarMap.get(similar)+val.get());
            } else {
                similarMap.put(similar, val.get());
            }
//            System.out.println(key.toString() + " === " + similarMap.toString());
        }
        
        List<Map.Entry<Integer, Integer>> list=new ArrayList<Map.Entry<Integer, Integer>>(similarMap.entrySet());
        
        Collections.sort(list,new Comparator<Map.Entry<Integer, Integer>>(){
            public int compare(Map.Entry<Integer, Integer> obj1, 
                    Map.Entry<Integer, Integer> obj2){
                int compare = Integer.compare(obj2.getValue(), obj1.getValue());
                if(compare == 0) {
                    compare = Integer.compare(obj1.getKey(), obj2.getKey());
                }
                return compare;
            }
        });
        
        StringBuffer sb = new StringBuffer("");
        
        List<Map.Entry<Integer, Integer>> outputList = list.size() > 10 ?
                                                        list.subList(0, 10): list;
        
        for (Map.Entry<Integer, Integer> entry : outputList) {
            sb.append(entry.getKey() + " ");
        }
        
        context.write(new Text(key.getUser()+":"), new Text(sb.toString()));
        
    } 
    
}
